﻿using JESAI.DistributedLock.Internal;
using StackExchange.Redis;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace JESAI.DistributedLock.Redis.RedLock;

internal interface IRedLockAcquirableSynchronizationPrimitive : IRedLockReleasableSynchronizationPrimitive
{
    TimeoutValue AcquireTimeout { get; }
    Task<bool> TryAcquireAsync(IDatabaseAsync database);
    bool TryAcquire(IDatabase database);
    bool IsConnected(IDatabase database);
}

/// <summary>
/// 在RedLock算法中实现获取操作。请参阅 https://redis.io/topics/distlock
/// </summary>
internal readonly struct RedLockAcquire
{
    private readonly IRedLockAcquirableSynchronizationPrimitive _primitive;
    private readonly IReadOnlyList<IDatabase> _databases;
    private readonly CancellationToken _cancellationToken;

    public RedLockAcquire(
        IRedLockAcquirableSynchronizationPrimitive primitive,
        IReadOnlyList<IDatabase> databases,
        CancellationToken cancellationToken)
    {
        this._primitive = primitive;
        this._databases = databases;
        this._cancellationToken = cancellationToken;
    }

    public async ValueTask<Dictionary<IDatabase, Task<bool>>?> TryAcquireAsync()
    {
        this._cancellationToken.ThrowIfCancellationRequested();

        var isSynchronous = SyncViaAsync.IsSynchronous;
        if (isSynchronous && this._databases.Count == 1)
        {
            return this.TrySingleFullySynchronousAcquire();
        }

        var primitive = this._primitive;
        var tryAcquireTasks = this._databases.ToDictionary(
            db => db,
            db => Helpers.SafeCreateTask(state => state.primitive.TryAcquireAsync(state.db), (primitive, db))
        );

        var waitForAcquireTask = this.WaitForAcquireAsync(tryAcquireTasks);

        var succeeded = false;
        try
        {
            succeeded = await waitForAcquireTask.AwaitSyncOverAsync().ConfigureAwait(false);
        }
        finally
        {
            // 清理
            if (!succeeded)
            {
                List<Task>? releaseTasks = null;
                foreach (var kvp in tryAcquireTasks)
                {
                    // 如果任务还没有完成，我们现在不想做任何发布；只是将释放命令排队，以便在任务最终完成时运行
                    if (!kvp.Value.IsCompleted)
                    {
                        RedLockHelper.FireAndForgetReleaseUponCompletion(primitive, kvp.Key, kvp.Value);
                    }
                    // 否则，除非我们知道我们未能获得，否则就发布
                    else if (!RedLockHelper.ReturnedFalse(kvp.Value))
                    {
                        if (isSynchronous)
                        {
                            try { primitive.Release(kvp.Key, fireAndForget: true); }
                            catch { }
                        }
                        else
                        {
                            (releaseTasks ??= [])
                                .Add(Helpers.SafeCreateTask(state => state.primitive.ReleaseAsync(state.Key, fireAndForget: true), (primitive, kvp.Key)));
                        }
                    }
                }

                if (releaseTasks != null)
                {
                    await Task.WhenAll(releaseTasks).ConfigureAwait(false);
                }
            }
        }

        return succeeded ? tryAcquireTasks : null;
    }

    private async Task<bool> WaitForAcquireAsync(IReadOnlyDictionary<IDatabase, Task<bool>> tryAcquireTasks)
    {
        using var timeout = new TimeoutTask(this._primitive.AcquireTimeout, this._cancellationToken);
        var incompleteTasks = new HashSet<Task>(tryAcquireTasks.Values) { timeout.Task };

        var successCount = 0;
        var failCount = 0;
        var faultCount = 0;
        while (true)
        {
            var completed = TryResolveDisconnectedDatabaseAsFaulted(this)
                ?? await Task.WhenAny(incompleteTasks).ConfigureAwait(false);

            if (completed == timeout.Task)
            {
                await completed.ConfigureAwait(false); // 传播取消
                return false; // 真超时
            }

            if (completed.Status == TaskStatus.RanToCompletion)
            {
                var result = await ((Task<bool>)completed).ConfigureAwait(false);
                if (result)
                {
                    ++successCount;
                    if (RedLockHelper.HasSufficientSuccesses(successCount, this._databases.Count)) { return true; }
                }
                else
                {
                    ++failCount;
                    if (RedLockHelper.HasTooManyFailuresOrFaults(failCount, this._databases.Count)) { return false; }
                }
            }
            else // 故障或取消
            {
                // 如果我们犯了太多错误，就不可能获得锁，所以我们应该抛出
                ++faultCount;
                if (RedLockHelper.HasTooManyFailuresOrFaults(faultCount, this._databases.Count))
                {
                    var faultingTasks = tryAcquireTasks.Values.Where(t => t.IsCanceled || t.IsFaulted)
                        .ToArray();
                    if (faultingTasks.Length == 1)
                    {
                        await faultingTasks[0].ConfigureAwait(false); // propagate the error
                    }

                    throw new AggregateException(faultingTasks.Select(t => t.Exception ?? new TaskCanceledException(t).As<Exception>()))
                        .Flatten();
                }

                ++failCount;
                if (RedLockHelper.HasTooManyFailuresOrFaults(failCount, this._databases.Count)) { return false; }
            }

            incompleteTasks.Remove(completed);
            Invariant.Require(incompleteTasks.Count > 1, "should be more than just timeout left");
        }

        // MA:这种行为不平凡，值得详细解释。基本上是StackExchange。Redis 2.5.27在对断开连接的数据库触发命令时切换了行为，这样这些命令就会积压一段时间
        // 等待重新连接，而不是快速失败（快速失败仍然是ConnectionMultiplexer上的非默认选项，但我们不想要求这样）。
        //
        // 虽然这种行为通常是有道理的，但它会给RedLock算法带来长时间的延迟。例如，想象一下，当C关闭时，进程1和2锁定在服务器a、B和C上的情况。如果进程1在A上
        // 获取，进程2在B上获取，则C将投出获胜的一票，为此，两个线程都必须等待完整的连接超时来观察问题并宣布获取失败。
        //
        // 由此造成的延迟在我们的TestParallelism（）案例中非常明显，该案例模拟了2x1场景中宕机的服务器：它们的严重程度足以使测试失败！
        //
        // 我实现的修复是，在多服务器场景中，当服务器投出“决定性投票”时，通过连接检查绕过积压。这样，上述情况就会很快得到解决，因为这两个过程都会看到C断
        // 开连接，导致当前获取失败而无需等待。请注意，在单服务器场景中，这始终是不允许的。
        //
        // 当前的方法无法处理决定性投票被多个服务器关闭的情况。我们可以以增加复杂性为代价来实现这一点，但目前我认为这是不值得的，因为无论如何，这种情况应
        // 该不那么常见，问题也会更多。
        //
        // 最后，请注意，虽然这种行为可以扩展到所有RedLock操作（扩展、释放），但我认为这没有价值，因为在正常情况下只有acquire会受到争用，所以优化在其他地
        // 方没有任何价值。
        //
        // 相关链接：
        // https://github.com/madelson/DistributedLock/pull/173#discussion_r1342236087
        // https://github.com/StackExchange/StackExchange.Redis/issues/2645
        Task? TryResolveDisconnectedDatabaseAsFaulted(RedLockAcquire @this)
        {
            // 首先，检查（a）我们是否至少有一次成功/失败，以及（b）是否还有一次是决定性的。如果没有，保释。
            if (!((successCount > 0 && RedLockHelper.HasSufficientSuccesses(successCount + 1, @this._databases.Count))
                || (failCount > 0 && RedLockHelper.HasTooManyFailuresOrFaults(failCount + 1, @this._databases.Count))))
            {
                return null;
            }

            // 迭代任务，查看是否所有未完成的任务都已断开连接。如果是这样，请选择一个来解决
            Task? toResolve = null;
            foreach (var kvp in tryAcquireTasks)
            {
                if (incompleteTasks.Contains(kvp.Value))
                {
                    if (kvp.Value.IsCompleted)
                    {
                        return null;
                    }
                    if (toResolve is null && !@this._primitive.IsConnected(kvp.Key))
                    {
                        toResolve = kvp.Value;
                        // 不要回到这里，因为如果另一个任务完成了，我们希望它优先
                    }
                }
            }

            if (toResolve is null) { return null; }

            // 在此处删除此项，因为我们将用已解决的任务替换它，以便稍后调用completeTasks。Remove（）将不起作用
            incompleteTasks.Remove(toResolve);
            return Task.FromException(new RedisException("Database is disconnected"));
        }
    }

    /// <summary>
    /// 由于StackExchange，我们只允许对单个数据库进行同步获取。Redis目前不允许单次操作超时/取消。因此，一个缓慢的响应将危及我们及时申请锁的能力。
    /// 对于单个数据库，一个操作才是最重要的，所以如果我们需要等待它，那也没关系。
    /// </summary>
    private Dictionary<IDatabase, Task<bool>>? TrySingleFullySynchronousAcquire()
    {
        var database = this._databases.Single();

        bool success;
        var stopwatch = Stopwatch.StartNew();
        try { success = this._primitive.TryAcquire(database); }
        catch
        {
            // 如果失败，仍然尝试释放，以防万一
            try { this._primitive.Release(database, fireAndForget: true); }
            catch { } // 什么都不做；不管怎样，我们还是要扔，失败的原因可能是一样的

            throw;
        }

        if (success)
        {
            // 确保我们没有超时
            if (this._primitive.AcquireTimeout.CompareTo(stopwatch.Elapsed) >= 0)
            {
                return new Dictionary<IDatabase, Task<bool>> { [database] = Task.FromResult(success) };
            }

            this._primitive.Release(database, fireAndForget: true); // 已超时，请释放
        }

        return null;
    }
}
